#pragma once
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"

#include "./connection.hpp"
#include "./consumer.hpp"
#include "./virtualhost.hpp"
#include "../mqcommon/threadpool.hpp"
#include "../mqcommon/msg.pb.h"
#include "../mqcommon/proto.pb.h"

#ifndef __BROKER_HOLDERS__
#define __BROKER_HOLDERS__
#define _1 std::placeholders::_1
#define _2 std::placeholders::_2
#define _3 std::placeholders::_3
namespace mymq
{
    class BrokerServer
    {
    public:

    public:
        BrokerServer(const uint16_t port, const std::string &hostname,
                        const std::string &basedir, const std::string& dbfile)
        : _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "Server", muduo::net::TcpServer::kReusePort)
        , _dispatcher(std::bind(&BrokerServer::OnUnknownMessage, this, _1, _2, _3))
        , _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, _1, _2, _3)))
        , _host(std::make_shared<VirtualHost>(hostname, basedir, dbfile))
        , _consumer_manager(std::make_shared<ConsumerManager>())
        , _connection_manager(std::make_shared<ConnectionManager>())
        , _pool(std::make_shared<ThreadPool>())
        {
            // 针对历史消息中的所有队列，别忘了，初始化队列的消费者结构
            auto queuemap = _host->GetAllMsgQueue();
            for (const auto& queue: queuemap)
            {
                _consumer_manager->InitQueueConsumer(queue);
            }
            // 注册业务处理函数
            _dispatcher.registerMessageCallback<mymq::OpenChannelRequest>(
                std::bind(&BrokerServer::OnCreateChannel, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<mymq::CloseChannelRequest>(
                std::bind(&BrokerServer::OnDestoryChannel, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<mymq::DeclareExchangeRequest>(
                std::bind(&BrokerServer::OnDeclareExchange, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<mymq::DeleteExchangeRequest>(
                std::bind(&BrokerServer::OnDeleteExchange, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<mymq::DeclareQueueRequest>(
                std::bind(&BrokerServer::OnDeclareQueue, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<mymq::QueueBindRequest>(
                std::bind(&BrokerServer::OnQueueBind, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<mymq::QueueUnBindRequest>(
                std::bind(&BrokerServer::OnQueueUnBind, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<mymq::PublishRequest>(
                std::bind(&BrokerServer::OnPublish, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<mymq::AckRequest>(
                std::bind(&BrokerServer::OnAck, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<mymq::ConsumeRequest>(
                std::bind(&BrokerServer::OnConsume, this, _1, _2, _3));
            _dispatcher.registerMessageCallback<mymq::CancelRequest>(
                std::bind(&BrokerServer::OnCancel, this, _1, _2, _3));
            _server.setConnectionCallback(
                std::bind(&BrokerServer::OnConnection, this, _1));
            _server.setMessageCallback(
                std::bind(&ProtobufCodec::onMessage, _codec.get(), _1, _2, _3));
        }
        void Start()
        {
            _server.start();
            _baseloop.loop();
        }
    private:
        void OnCreateChannel(const muduo::net::TcpConnectionPtr& conn,
                const OpenChannelRequestPtr& message,
                muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->GetConnection(conn);
            if (myconn.get() == nullptr)
            {
                INFO_LOG("打开信道时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            myconn->CreateChannel(message);
        }
        void OnDestoryChannel(const muduo::net::TcpConnectionPtr& conn,
                const CloseChannelRequestPtr& message,
                muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->GetConnection(conn);
            if (myconn.get() == nullptr)
            {
                INFO_LOG("关闭信道时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            myconn->DestoryChannel(message);
        }
        void OnDeclareExchange(const muduo::net::TcpConnectionPtr& conn,
                const DeclareExchangeRequestPtr& message,
                muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->GetConnection(conn);
            if (myconn.get() == nullptr)
            {
                INFO_LOG("声明交换机时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = myconn->SelectChannel(message->cid());
            if (channel.get() == nullptr)
            {
                INFO_LOG("声明交换机时，没有找到连接对应的信道");
                return;
            }
            channel->DeclareExchange(message);
        }
        void OnDeleteExchange(const muduo::net::TcpConnectionPtr& conn,
                const DeleteExchangeRequestPtr& message,
                muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->GetConnection(conn);
            if (myconn.get() == nullptr)
            {
                INFO_LOG("删除交换机时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = myconn->SelectChannel(message->cid());
            if (channel.get() == nullptr)
            {
                INFO_LOG("删除交换机时，没有找到连接对应的信道");
                return;
            }
            channel->DeleteExchange(message);
        }
        void OnDeclareQueue(const muduo::net::TcpConnectionPtr& conn,
                const DeclareQueueRequestPtr& message,
                muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->GetConnection(conn);
            if (myconn.get() == nullptr)
            {
                INFO_LOG("声明队列时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = myconn->SelectChannel(message->cid());
            if (channel.get() == nullptr)
            {
                INFO_LOG("声明队列时，没有找到连接对应的信道");
                return;
            }
            channel->DeclareQueue(message);
        }
        void OnDeleteQueue(const muduo::net::TcpConnectionPtr& conn,
                const DeleteQueueRequestPtr& message,
                muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->GetConnection(conn);
            if (myconn.get() == nullptr)
            {
                INFO_LOG("删除队列时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = myconn->SelectChannel(message->cid());
            if (channel.get() == nullptr)
            {
                INFO_LOG("删除队列时，没有找到连接对应的信道");
                return;
            }
            channel->DeleteQueue(message);
        }
        void OnQueueBind(const muduo::net::TcpConnectionPtr& conn,
                const QueueBindRequestPtr& message,
                muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->GetConnection(conn);
            if (myconn.get() == nullptr)
            {
                INFO_LOG("队列绑定时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = myconn->SelectChannel(message->cid());
            if (channel.get() == nullptr)
            {
                INFO_LOG("队列绑定时，没有找到连接对应的信道");
                return;
            }
            channel->QueueBind(message);
        }
        void OnQueueUnBind(const muduo::net::TcpConnectionPtr& conn,
                const QueueUnBindRequestPtr& message,
                muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->GetConnection(conn);
            if (myconn.get() == nullptr)
            {
                INFO_LOG("队列解绑时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = myconn->SelectChannel(message->cid());
            if (channel.get() == nullptr)
            {
                INFO_LOG("队列解绑时，没有找到连接对应的信道");
                return;
            }
            channel->QueueUnBind(message);
        }
        void OnPublish(const muduo::net::TcpConnectionPtr& conn,
                const PublishRequestPtr& message,
                muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->GetConnection(conn);
            if (myconn.get() == nullptr)
            {
                INFO_LOG("消息发布时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = myconn->SelectChannel(message->cid());
            if (channel.get() == nullptr)
            {
                INFO_LOG("消息发布时，没有找到连接对应的信道");
                return;
            }
            channel->Publish(message);
        }
        void OnAck(const muduo::net::TcpConnectionPtr& conn,
                const AckRequestPtr& message,
                muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->GetConnection(conn);
            if (myconn.get() == nullptr)
            {
                INFO_LOG("消息确认时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = myconn->SelectChannel(message->cid());
            if (channel.get() == nullptr)
            {
                INFO_LOG("消息确认时，没有找到连接对应的信道");
                return;
            }
            channel->Ack(message);
        }
        void OnConsume(const muduo::net::TcpConnectionPtr& conn,
                const ConsumeRequestPtr& message,
                muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->GetConnection(conn);
            if (myconn.get() == nullptr)
            {
                INFO_LOG("队列消息订阅时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = myconn->SelectChannel(message->cid());
            if (channel.get() == nullptr)
            {
                INFO_LOG("队列消息订阅时，没有找到连接对应的信道");
                return;
            }
            channel->Consume(message);
        }
        void OnCancel(const muduo::net::TcpConnectionPtr& conn,
                const CancelRequestPtr& message,
                muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->GetConnection(conn);
            if (myconn.get() == nullptr)
            {
                INFO_LOG("取消订阅时，没有找到连接对应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = myconn->SelectChannel(message->cid());
            if (channel.get() == nullptr)
            {
                INFO_LOG("取消订阅时，没有找到连接对应的信道");
                return;
            }
            channel->Cancel(message);
        }
        void OnConnection(const muduo::net::TcpConnectionPtr& conn)
        {
            if (conn->connected())
            {
                LOG_INFO << "新连接成功\n";
                _connection_manager->NewConnection(_consumer_manager, 
                _host, conn, _codec, _pool);
            }
            else
            {
                LOG_INFO << "连接断开\n";
                _connection_manager->DeleteConnection(conn);
            }
        }
        void OnUnknownMessage(const muduo::net::TcpConnectionPtr& conn,
                            const ::MessagePtr& message,
                            muduo::Timestamp)
        {
            // const ::MessagePtr& message 是为了用codec.h中全局的MessagePtr
            LOG_INFO << "OnUnknownMessage: " << message->GetTypeName();
            conn->shutdown();
        }
    private:
        muduo::net::EventLoop _baseloop;
        muduo::net::TcpServer _server; // tcp服务器对象
        ProtobufDispatcher _dispatcher; // 分发器
        ProtobufCodecPtr _codec; //协议器
        VirtualHost::ptr _host;
        ConsumerManager::ptr _consumer_manager;
        ConnectionManager::ptr _connection_manager;
        ThreadPool::ptr _pool;
    };
}
#endif